package org.yuanzheng.sink

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer

/**
 * @author yuanzheng
 * @date 2020/6/16-21:02
 */
object KafkaStringSink {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    //读取数据源
    val stream: DataStream[String] = streamEnv.socketTextStream("192.168.1.9", 8888)
    //处理
    val words: DataStream[String] = stream.flatMap(_.split(" "))
    //把单词写入kafka
    words.addSink(new FlinkKafkaProducer[String]("192.168.1.8:9092", "yuan", new SimpleStringSchema()))
    streamEnv.execute()
  }
}
